/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.xhy.sniffer.datacarrier.consumer;


import org.xhy.sniffer.datacarrier.buffer.Channels;
import org.xhy.sniffer.datacarrier.buffer.QueueBuffer;

import java.util.ArrayList;
import java.util.List;

/**
 * 多个 channel
 * IConsumer}
 */
public class MultipleChannelsConsumer extends Thread {
    private volatile boolean running; // 用于指示线程是否在运行，volatile 保证线程安全的状态读写
    private volatile ArrayList<Group> consumeTargets; // 保存所有要消费的目标（即通道和消费者）
    @SuppressWarnings("NonAtomicVolatileUpdate")
    private volatile long size; // 当前消费目标的总大小
    private final long consumeCycle; // 消费周期，表示线程在无数据可消费时的休眠时间

    // 构造函数，初始化线程名称和消费周期
    public MultipleChannelsConsumer(String threadName, long consumeCycle) {
        super(threadName); // 设置线程名称
        this.consumeTargets = new ArrayList<Group>(); // 初始化消费目标列表
        this.consumeCycle = consumeCycle; // 设置消费周期
    }

    @Override
    public void run() {
        running = true; // 设置线程为运行状态

        final List consumeList = new ArrayList(2000); // 用于存放从缓冲区获取到的数据，初始容量为2000
        while (running) { // 当线程处于运行状态时
            boolean hasData = false; // 标记当前循环中是否有数据被消费
            for (Group target : consumeTargets) { // 遍历所有消费目标
                boolean consume = consume(target, consumeList); // 尝试消费数据
                hasData = hasData || consume; // 如果有数据被消费，设置标志
            }

            if (!hasData) { // 如果没有数据被消费，线程进入休眠
                try {
                    Thread.sleep(consumeCycle); // 休眠指定的消费周期
                } catch (InterruptedException e) {
                    // 捕获异常，但不进行处理
                }
            }
        }

        // 线程准备停止时，最后一次消费所有目标的数据
        for (Group target : consumeTargets) {
            consume(target, consumeList); // 消费数据

            target.consumer.onExit(); // 调用消费者的退出方法
        }
    }

    // 消费指定目标的数据
    private boolean consume(Group target, List consumeList) {
        for (int i = 0; i < target.channels.getChannelSize(); i++) { // 遍历目标的所有通道
            QueueBuffer buffer = target.channels.getBuffer(i); // 获取当前通道的缓冲区
            buffer.obtain(consumeList); // 从缓冲区获取数据并添加到 consumeList
        }

        if (!consumeList.isEmpty()) { // 如果获取到了数据
            try {
                target.consumer.consume(consumeList); // 调用消费者的 consume 方法处理数据
            } catch (Throwable t) {
                target.consumer.onError(consumeList, t); // 如果消费过程中发生异常，调用 onError 处理
            } finally {
                consumeList.clear(); // 清空消费列表，准备下一次消费
            }
            return true; // 返回 true，表示有数据被消费
        }

        target.consumer.nothingToConsume(); // 如果没有数据，调用 nothingToConsume 方法
        return false; // 返回 false，表示没有数据被消费
    }

    /**
     * 添加新的消费目标。
     */
    public void addNewTarget(Channels channels, IConsumer consumer) {
        Group group = new Group(channels, consumer); // 创建一个新的 Group 包含通道和消费者
        // 创建一个新的列表，以避免在消费时修改列表,并发问题
        ArrayList<Group> newList = new ArrayList<Group>();
        for (Group target : consumeTargets) { // 将现有的消费目标添加到新列表中
            newList.add(target);
        }
        newList.add(group); // 添加新的目标
        consumeTargets = newList; // 更新消费目标列表
        size += channels.size(); // 更新总大小
    }

    // 返回当前消费目标的总大小
    public long size() {
        return size;
    }

    // 关闭消费者线程
    void shutdown() {
        running = false; // 设置运行状态为 false，线程将在下一次循环时停止
    }

    // 静态内部类，用于将通道和消费者组合为一个消费目标
    private static class Group {
        private Channels channels; // 通道，包含数据缓冲区
        private IConsumer consumer; // 消费者，负责处理从通道中获取的数据

        // 构造函数，初始化通道和消费者
        public Group(Channels channels, IConsumer consumer) {
            this.channels = channels;
            this.consumer = consumer;
        }
    }
}

